package cn.fz.rocketmqdemo01.service.impl;

import cn.fz.rocketmqdemo01.service.RocketAdminService;
import cn.fz.rocketmqdemo01.utils.RocketmqAdminFactory;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
public class RocketAdminServiceImpl implements RocketAdminService {
    @Autowired
    private RocketmqAdminFactory rocketmqAdminFactory;

    public ClusterInfo getClusterInfo(String serverAddress){
        try {
            return rocketmqAdminFactory.getOrCreateAdmin(serverAddress).examineBrokerClusterInfo();
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<String> getClusterNameList(String serverAddress) {
        ClusterInfo clusterInfo = getClusterInfo(serverAddress);
        return new ArrayList<>(clusterInfo.getClusterAddrTable().keySet());
    }

    @Override
    public List<String> getTopicList(String serverAddress) {
        try {
            TopicList topicList = rocketmqAdminFactory.getOrCreateAdmin(serverAddress).fetchAllTopicList();
            return new ArrayList<>(topicList.getTopicList());
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<BrokerData> getBrokerDataList(String serverAddress) {
        ClusterInfo clusterInfo = getClusterInfo(serverAddress);
        HashMap<String, BrokerData> brokerAddrTable = (HashMap<String, BrokerData>) clusterInfo.getBrokerAddrTable();
        return new ArrayList<>(brokerAddrTable.values());
    }

    @Override
    public void createTopic(String serverAddress, String topicName) {
        try {
            MQAdminExt admin = rocketmqAdminFactory.getOrCreateAdmin(serverAddress);
            // 创建主题配置对象
            TopicConfig topicConfig = new TopicConfig();
            topicConfig.setTopicName(topicName);
            topicConfig.setPerm(6);
            topicConfig.setOrder(false); // 非顺序消息
            topicConfig.setReadQueueNums(16); // 读取队列数量
            topicConfig.setWriteQueueNums(16); // 写入队列数量
            // 在指定的Broker上创建主题
            for (BrokerData brokerData : getBrokerDataList(serverAddress)) {
                admin.createAndUpdateTopicConfig(brokerData.selectBrokerAddr(), topicConfig);
            }
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void deleteTopic(String serverAddress, String topicName) {
        try {
            MQAdminExt admin = rocketmqAdminFactory.getOrCreateAdmin(serverAddress);
            admin.deleteTopicInNameServer(new HashSet<>(Collections.singletonList(serverAddress)), topicName);
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }
}
